package com.atguigu.edu.realtime.app.dws;

import com.atguigu.edu.realtime.beans.KeywordBean;
import com.atguigu.edu.realtime.func.KeywordUDTF;
import com.atguigu.edu.realtime.utils.MyClickhouseUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //TODO 2 检查点相关设置
        /*//2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/edu/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 3 从kafka的page_log主题中读取页面日志-创建动态表指定Watermark以及提取事件时间字段
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_keyword_group";
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "  common map<string,string>,\n" +
                "  page map<string,string>,\n" +
                "  ts BIGINT,\n" +
                "  row_time as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
                "  WATERMARK FOR row_time AS row_time - INTERVAL '3' SECOND\n" +
                ")" + MyKafkaUtil.getKafkaDDL(topic,groupId));
        //tableEnv.executeSql("select * from page_log").print();
        //TODO 4 过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
                "\tpage['item'] keyword,\n" +
                "\trow_time\n" +
                "from page_log\n" +
                "where page['item'] is not null and page['item_type'] = 'keyword'");
        tableEnv.createTemporaryView("search_table",searchTable);
        //tableEnv.executeSql("select * from search_table").print();
        //TODO 5 使用自定义函数对搜索的内容进行分词 将原表字段和表函数执行的结果进行连接 本项目keyword不需要分词，跳过
        //TODO 6 分组开窗聚合计算
        Table aggTable = tableEnv.sqlQuery("select \n" +
                "\tDATE_FORMAT(TUMBLE_START(row_time, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "    DATE_FORMAT(TUMBLE_END(row_time, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') edt,\n" +
                "    keyword,\n" +
                "    count(*) keyword_rate,\n" +
                "    UNIX_TIMESTAMP()*1000 ts\n" +
                "from search_table group by TUMBLE(row_time, INTERVAL '10' SECOND) ,keyword");
        //tableEnv.createTemporaryView("agg_table",aggTable);
        //tableEnv.executeSql("select * from agg_table").print();
        //TODO 7 将动态表转换成流
        DataStream<KeywordBean> keywordDS = tableEnv.toAppendStream(aggTable, KeywordBean.class);
        keywordDS.print(">>>");
        //TODO 8.将流中的数据写到CK中
        keywordDS.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?)")
        );

        env.execute();
    }
}
